package org.example

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import java.util

object kafkaWordProducer {
  def main(args: Array[String]): Unit = {
    var brokers="192.168.152.121:9092"
    var topic="spark12"
    var messagePerSec=3
    var wordPerMessage=5
    var props=new util.HashMap[String,Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
    val producer=new KafkaProducer[String,String](props)
    while(true){
      (1 to messagePerSec).foreach{messagenum=>
        var str=(1 to wordPerMessage).map(x=>scala.util.Random.nextInt(10).toString).mkString(" ")
        print(str)
        println()
        var message=new ProducerRecord[String,String](topic,null,str)
        producer.send(message)
      }
      Thread.sleep(1000)
    }
  }
}
